package cn.yunyichina.provider.pay.executor;

import cn.yunyichina.provider.framework.data.SpringContextHandler;
import cn.yunyichina.provider.framework.kafka.KafkaTopic;
import cn.yunyichina.provider.framework.utils.UUIDUtil;
import cn.yunyichina.provider.iface.entity.base.OrderExtVo;
import cn.yunyichina.provider.iface.entity.pay.MchData;
import cn.yunyichina.provider.iface.entity.pay.PayCode;
import cn.yunyichina.provider.iface.entity.pay.notify.UnifiedPayNotifyMessage;
import cn.yunyichina.provider.pay.constant.InsuranceMode;
import cn.yunyichina.provider.pay.constant.Medicare;
import cn.yunyichina.provider.pay.dao.PaymentJournalDao;
import cn.yunyichina.provider.pay.entity.PaymentJournal;
import cn.yunyichina.provider.pay.service.BaseService;
import cn.yunyichina.provider.pay.utils.NumberUtil;
import cn.yunyichina.utils.commons.DateUtils;
import cn.yunyichina.utils.convert.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.math.BigDecimal;

/**
 * 支付通知业务处理线程类
 * <p>
 * Created by Lullaby on 2017/7/21
 */
public class PayNotifyWorker extends BaseService implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(PayNotifyWorker.class);

    private static KafkaTemplate<Integer, String> kafkaTemplate = SpringContextHandler.getBean(KafkaTemplate.class);

    private static PaymentJournalDao paymentJournalDao = SpringContextHandler.getBean(PaymentJournalDao.class);

    private UnifiedPayNotifyMessage unifiedMessage;

    private String originContent;

    public PayNotifyWorker(UnifiedPayNotifyMessage unifiedMessage) {
        this.unifiedMessage = unifiedMessage;
    }

    public PayNotifyWorker(UnifiedPayNotifyMessage unifiedMessage, String originContent) {
        this.unifiedMessage = unifiedMessage;
        this.originContent = originContent;
    }

    public void run() {
        logger.info("PayNotifyWorker.run.unifiedMessage -> {}", JsonUtils.toJsonString(unifiedMessage));

        // 1.产出消息
        try {
            MchData mchData = unifiedMessage.getMchData();

            if (mchData != null && !StringUtils.isEmpty(mchData.getTopic())) {
                String topic = mchData.getTopic();

                switch (KafkaTopic.valueOf(topic)) {
                    case REGISTER_ALIPAY_PAY:
                        produceMessage(topic, JsonUtils.toJsonString(unifiedMessage));
                        break;
                    case REGISTER_PAY_NOTIFY:
                        produceMessage(topic, JsonUtils.toJsonString(unifiedMessage));
                        break;
                    case CLINIC_PAY_NOTIFY:
                        produceMessage(topic, JsonUtils.toJsonString(unifiedMessage));
                        break;
                    case DRUG_ALIPAY_PAY:
                        produceMessage(topic, JsonUtils.toJsonString(unifiedMessage));
                        break;
                    default:
                        produceMessage(topic, originContent);
                        break;
                }
            }
        } catch (Exception e) {
            logger.error("PayNotifyWorker.run.produceMessage.Exception", e);
        }

        // 2.入流水(写入云医支付流水)
        try {
            paymentJournalDao.insert(buildPaymentJournal());
        } catch (Exception e) {
            logger.error("PayNotifyWorker.run.insertPaymentJournal.Exception", e);
        }
    }

    /**
     * 构建支付流水对象
     *
     * @return
     */
    private PaymentJournal buildPaymentJournal() {
        OrderExtVo orderExtVo = this.getBaseOrderByOrderNo(unifiedMessage.getOrderNo());

        PaymentJournal journal = new PaymentJournal();

        journal.setId(UUIDUtil.get32UUID());
        journal.setPaymentDealNo(NumberUtil.buildPaymentDealNo());

        journal.setOrgId(orderExtVo.getOrgId());
        journal.setOrgCode(orderExtVo.getOrgCode());
        journal.setOrgName(orderExtVo.getOrgName());
        journal.setPlatformId(orderExtVo.getPlatformId());
        journal.setPlatformName(orderExtVo.getPlatformName());
        journal.setPlatformOrgAppCode(orderExtVo.getPlatformOrgAppCode());
        journal.setPlatformOrgAppName(orderExtVo.getPlatformOrgAppName());
        journal.setBranchId(orderExtVo.getBranchId());
        journal.setBranchCode(orderExtVo.getBranchCode());
        journal.setBranchName(orderExtVo.getBranchName());
        journal.setOutOrderNo(orderExtVo.getOutOrderNo());
        journal.setOrderType(orderExtVo.getOrderType());

        journal.setPayAmount(new BigDecimal(unifiedMessage.getPayAmount()).intValue());

        String insuranceAmount = unifiedMessage.getInsuranceAmount();
        if (StringUtils.isEmpty(insuranceAmount) || insuranceAmount.equals("0")) {
            insuranceAmount = orderExtVo.getInsuranceAmount() == null ? "0" : String.valueOf(orderExtVo.getInsuranceAmount());
        }

        if (!StringUtils.isEmpty(insuranceAmount)) {
            journal.setAccountAmount(new BigDecimal(insuranceAmount).intValue());
            journal.setMedicareAmount(orderExtVo.getMedicareAmount());
            journal.setInsuranceAmount(new BigDecimal(insuranceAmount).intValue());
        } else {
            journal.setAccountAmount(orderExtVo.getAccountAmount());
            journal.setMedicareAmount(orderExtVo.getMedicareAmount());
            journal.setInsuranceAmount(orderExtVo.getInsuranceAmount() == null ? 0 : orderExtVo.getInsuranceAmount());
        }

        String totalAmount = unifiedMessage.getTotalAmount();
        if (StringUtils.isEmpty(totalAmount)) {
            journal.setTotalAmount(journal.getPayAmount() + journal.getInsuranceAmount());
        } else {
            journal.setTotalAmount(new BigDecimal(totalAmount).intValue());
        }

        MchData mchData = unifiedMessage.getMchData();
        String isInsurance = mchData.getIsInsurance();
        String payCode = mchData.getPayCode();
        if (!StringUtils.isEmpty(isInsurance) && Medicare.INSURANCE_MODE_YEP.equals(isInsurance)) {
            switch (PayCode.valueOf(payCode)) {
                case ALIPAYMEDICARE:
                    journal.setInsuranceMode(Short.valueOf(String.valueOf(InsuranceMode.ALIPAY.getCode())));
                    break;
                case WXMEDICALPAY:
                    journal.setInsuranceMode(Short.valueOf(String.valueOf(InsuranceMode.WECHAT.getCode())));
                    break;
                case PACITY:
                    journal.setInsuranceMode(Short.valueOf(String.valueOf(InsuranceMode.PACITY.getCode())));
                    break;
                default:
                    break;
            }
        } else {
            journal.setInsuranceMode(Short.valueOf(String.valueOf(InsuranceMode.OTHERS.getCode())));
        }

        journal.setPayAppId(unifiedMessage.getAppId());
        journal.setOrderNo(unifiedMessage.getOrderNo());
        journal.setHisOrderNo(orderExtVo.getHisOrderNo());
        journal.setPaymentDealId(unifiedMessage.getTradeNo());
        journal.setPayStatus((short) 1);
        journal.setCheckStatus((short) 0);
        journal.setSettlementStatus((short) 0);
        journal.setPayCode(payCode);
        journal.setPayMode((short) getPayModeByPayCode(payCode));

        journal.setDealAccount(unifiedMessage.getOpenId());
        journal.setPayTime(unifiedMessage.getPayTime());
        journal.setCreatedTime(DateUtils.getDatetime());

        logger.info("PayNotifyWorker.buildPaymentJournal.journal -> {}", JsonUtils.toJsonString(journal));

        return journal;
    }

    /**
     * 产出支付通知消息到消息队列(业务方监听消息完成支付闭环)
     *
     * @param topic
     * @param message
     */
    private void produceMessage(String topic, String message) {
        logger.info("PayNotifyWorker.produceMessage.topic & message -> {} & {}", topic, message);

        String kafkaTopic = KafkaTopic.valueOf(topic).getTopic();
        logger.info("PayNotifyWorker.produceMessage.kafkaTopic -> {}", kafkaTopic);

        ListenableFuture<SendResult<Integer, String>> listenableFuture = kafkaTemplate.send(kafkaTopic, message);

        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            public void onSuccess(SendResult<Integer, String> sendResult) {
                logger.info("PayNotifyWorker.produceMessage.onSuccess.getProducerRecord -> {}", sendResult.getProducerRecord().toString());
            }

            public void onFailure(Throwable throwable) {
                logger.error("PayNotifyWorker.produceMessage.onFailure", throwable);
            }
        });
    }

}
